package com.taobao.ad.hadoop.common;

import java.io.IOException;
import java.util.Iterator;
 
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter; 

public class MapReduce extends JobNode {
	
	public static class Map extends MapReduceBase implements Mapper<NullWritable, BytesWritable, BytesWritable, BytesWritable> {
		@Override
		public void map(NullWritable key, BytesWritable value, OutputCollector<BytesWritable, BytesWritable> output, Reporter reporter) throws IOException {
			// TODO Auto-generated method stub
			BytesWritable outKey = new BytesWritable();
			output.collect(outKey, value);
		}

		@Override
		public void configure(JobConf job) {
			super.configure(job);
		}
	}

	public static class Reduce extends MapReduceBase implements Reducer<BytesWritable, BytesWritable, NullWritable, BytesWritable> {
		@Override
		public void reduce(BytesWritable key, Iterator<BytesWritable> values, OutputCollector<NullWritable, BytesWritable> output, Reporter reporter) throws IOException {
			// TODO Auto-generated method stub
			while (values.hasNext()) {
				output.collect(NullWritable.get(), values.next());
			}
		}

		@Override
		public void configure(JobConf job) {
			super.configure(job);
		}
	}

	public static class Combine extends MapReduceBase implements Reducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable> {
		@Override
		public void reduce(BytesWritable key, Iterator<BytesWritable> values, OutputCollector<BytesWritable, BytesWritable> output, Reporter reporter) throws IOException {
			// TODO Auto-generated method stub
			while (values.hasNext()) {
				output.collect(key, values.next());
			}
		}

		@Override
		public void configure(JobConf job) {
			super.configure(job);
		}
	}

	public void ConfigJob() {
		super.ConfigJob(this.getClass(), Map.class, Reduce.class, null);
	}
}
